package com.xinsoft.config.mqtt;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.druid.util.StringUtils;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.xinsoft.config.SpringContext;
import com.xinsoft.constant.Const;
import com.xinsoft.entity.po.CusTestPoint;
import com.xinsoft.entity.po.CusWarnData;
import com.xinsoft.entity.po.CusWarnRule;
import com.xinsoft.entity.po.SysConfig;
import com.xinsoft.service.CusTestPointService;
import com.xinsoft.service.CusWarnDataService;
import com.xinsoft.service.CusWarnRuleService;
import com.xinsoft.service.SysConfigService;
import com.xinsoft.utils.InfluxDbUtils;
import com.xinsoft.utils.RedisUtil;
import com.xinsoft.utils.SpringUtils;
import com.xinsoft.websocket.WebSocketSessionHolderManager;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;

import javax.annotation.Resource;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/**
 * @Author Li9527
 * @Description 常规MQTT回调函数
 * @Date 2023/4/12 16:14
 **/
@Slf4j
public class Callback implements MqttCallback {

    private RedisUtil redisUtil = SpringContext.getBean(RedisUtil.class);

    private List<DeviceDataDO> list = new ArrayList<>();

    // 获取配置需告警时间service
    private SysConfigService configService = SpringContext.getBean(SysConfigService.class);

    // 测点service
    private CusTestPointService pointService = SpringContext.getBean(CusTestPointService.class);

    // 预警规则service
    private CusWarnRuleService warnRuleService = SpringContext.getBean(CusWarnRuleService.class);

    // 预警记录service
    private CusWarnDataService warnDataService = SpringContext.getBean(CusWarnDataService.class);

    // 满一千条的最后时间点
    private Long endTime = 0L;

    // 计时器，五分钟抽样一次
    private static Long dealTime = 0L;

    // websocket客户端
    private static WebSocketSessionHolderManager sessionHolderManager = SpringUtils.getBean(WebSocketSessionHolderManager.class);

    // 异步线程池丢inflush和发全局websocket消息
    public static ExecutorService create() {
        return ThreadUtil.newSingleExecutor();
    }

    /**
     * MQTT 断开连接会执行此方法
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.info("断开了MQTT连接 ：{}", throwable.getMessage());
        log.error(throwable.getMessage(), throwable);
    }

    /**
     * publish发布成功后会执行到这里
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
//        log.info("发布消息成功");
    }

//    @Override
//    public void connectionLost(Throwable cause) {
//
//    }

//    @Override
//    public void messageArrived(String topic, MqttMessage message) throws Exception {
//
//    }


    /**
     * subscribe订阅后得到的消息会执行到这里
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        //  TODO    此处可以将订阅得到的消息进行业务处理、数据存储
//        log.info("收到来自 " + topic + " 的消息：{}", new String(message.getPayload()));
        JSONObject msg = JSONObject.parseObject(new String(message.getPayload()));
        // 获取需告警时间
        Integer waringTime = 0;
        if (redisUtil.get(Const.POINT_OFFLINE_TIMEOUT) == null) {
            SysConfig sysConfig = configService.getOne(new QueryWrapper<SysConfig>().eq("param_key", Const.POINT_OFFLINE_TIMEOUT));
            redisUtil.set(Const.POINT_OFFLINE_TIMEOUT, sysConfig.getParamValue());
            waringTime = Integer.valueOf(sysConfig.getParamValue());
        } else {
            // 已存储
            waringTime = Integer.valueOf(redisUtil.get(Const.POINT_OFFLINE_TIMEOUT).toString());
        }
        ExecutorService executor = create();
        for (Map.Entry<String, Object> entry : msg.entrySet()) {
            DeviceDataDO deviceDataDO = JSONObject.parseObject(String.valueOf(msg), DeviceDataDO.class);
            String key = entry.getKey();
            if (key != "gatew" && key != "gateidc" && key != "time") {
                deviceDataDO.setPointId(key);
                deviceDataDO.setValue(entry.getValue().toString());
                deviceDataDO.setValueRecord(entry.getValue().toString());
                list.add(deviceDataDO);
                // 同时检索每个测点往redis里写一个带时效的key，如果key还未消失不操作，消失了再次写入
                if (redisUtil.get(Const.MONITOR_POINT + key) == null) {
                    redisUtil.set(Const.MONITOR_POINT + key, entry.getValue().toString(), waringTime);
                    executor.execute(() -> {
                        sessionHolderManager.sendBroadCast("网关" + deviceDataDO.getGateidc() + ";点位：" + deviceDataDO.getPointId() + "已超过配置掉线告警时长，请及时关注！");
                        newSaveInfluxData(list, "device_data");
                    });
                } else {
                    // 已存储
                    redisUtil.set(Const.MONITOR_POINT + key, entry.getValue().toString(), waringTime);
                }
            }
        }
        // todo 隔5分钟判断一次这批数据有没有需要预警的 先判断这个点位是否预警，有没有预警规则，然后对照上下限的值
        executor.execute(() -> {
            if (dealTime == 0) {
                dealTime = System.currentTimeMillis();
            }
            if ((System.currentTimeMillis() - dealTime) >= 1000 * 60 * 5) {
                dealTime = System.currentTimeMillis();
                if (list.size() > 0) {
                    for (int i = 0; i < list.size(); i++) {
                        // 取每个点位的预警配置值
                        CusTestPoint point = pointService.getOne(new QueryWrapper<CusTestPoint>().eq("test_point_code", list.get(i).getPointId()));
                        if (point.getDoWarn().equals(Const.Y)) {
                            CusWarnRule cusWarnRule = warnRuleService.getById(point.getWarnRuleId());
                            // 得到上下限,小于下限或大于上限了
                            if (BigDecimal.valueOf(Long.valueOf(list.get(i).getValue())).compareTo(cusWarnRule.getRuleMax()) != -1
                                    || BigDecimal.valueOf(Long.valueOf(list.get(i).getValue())).compareTo(cusWarnRule.getRuleMin()) != 1) {
                                CusWarnData warnData = new CusWarnData();
                                BeanUtil.copyProperties(point,warnData);
                                BeanUtil.copyProperties(cusWarnRule,warnData);
                                warnData.setNoticeContext("存在测点：" + point.getTestPointName() + "达到预警标准，当前值为：" + list.get(i).getValue()
                                        + ",允许的上限为：" + cusWarnRule.getRuleMax()  + ", 允许的下限为：" + cusWarnRule.getRuleMin());
                                warnDataService.save(warnData);
                            }
                        }
                    }
                }
            }
        });

        // 及时发送，但是持久化变更
        executor.execute(() -> {
            sessionHolderManager.sendBroadCast(JSONObject.toJSONString(list));
            newSaveInfluxData(list, "device_data");
            list.clear();
        });
        if (list.size() >= 100) {
            System.out.println("我已经连续积累了一百次网关数据,但是非生产环境，暂不存入influshDB");
            endTime = System.currentTimeMillis();
            executor.execute(() -> {
                newSaveInfluxData(list, "device_data");
                System.out.println("已持久化到influshDB一次" + ",当前暂存" + list.size() + "条");
            });
            // 如果监听到某个点位原来上传了数据，但是又间隔了配置预警时限未上传数据，及时告警。
            list.clear();
        } else {
            // 如果没满一千条，每三分钟持久化到influshDB一次
            if (System.currentTimeMillis() - endTime >= 1000 * 60 * 3) {
                executor.execute(() -> {
                    newSaveInfluxData(list, "device_data");
                });
            } else {
//                System.out.println("没满一千条，每三分钟持久化到influshDB一次" + ",当前已暂存" + list.size() + "条");
            }
        }
    }

    private void newSaveInfluxData(List<DeviceDataDO> deviceDatas, String measurement) {
        InfluxDbUtils influxDB = InfluxDbUtils.setUp();
        List<String> records = new ArrayList<>();
        for (DeviceDataDO dataMap : deviceDatas) {
            if (!StringUtils.isEmpty(dataMap.getValue().trim())) {
                try {
                    Map<String, String> tags = new HashMap<>();
                    tags.put("point", dataMap.getPointId());
                    tags.put("getWayId", dataMap.getGatew());
                    Map<String, Object> values = new HashMap<>();
                    values.put("value", Double.valueOf(Double.parseDouble(dataMap.getValue().toString().trim())));
                    values.put("valueRecord", Double.valueOf(Double.parseDouble(dataMap.getValueRecord().toString().trim())));
                    Point point = influxDB.pointBuilder(measurement, tags, values);
                    BatchPoints batchPoint = BatchPoints.database(measurement).consistency(InfluxDB.ConsistencyLevel.ALL).build();
                    batchPoint.point(point);
                    records.add(batchPoint.lineProtocol());
                } catch (Exception e) {
                    log.error(e.getMessage());
                }
            }
        }
        influxDB.batchInsert("device_data", null, InfluxDB.ConsistencyLevel.ALL, records);
    }

    private void newInfluxData(List<DeviceDataDO> deviceDatas, String measurement) {
        deviceDatas = (List<DeviceDataDO>) deviceDatas.stream().filter(e -> ("M310".equals(e.getPointId()) || "D130".equals(e.getPointId()))).collect(Collectors.toList());
        Set<String> getWayIds = (Set<String>) deviceDatas.stream().map(DeviceDataDO::getGateidc).collect(Collectors.toSet());
        List<RecombineData> recombineDatas = new ArrayList<>();
        for (String getWayId : getWayIds) {
            RecombineData recombineData1 = new RecombineData();
            for (DeviceDataDO deviceDataDO : deviceDatas) {
                if (getWayId.equals(deviceDataDO.getGateidc())) {
                    if ("M310".equals(deviceDataDO.getPointId())) {
                        recombineData1.setTime(deviceDataDO.getTime());
                        recombineData1.setPointId("1");
                        recombineData1.setGatewayNo(deviceDataDO.getGateidc());
                        recombineData1.setValue1(deviceDataDO.getValue());
                        continue;
                    }
                    recombineData1.setValue2(deviceDataDO.getValue());
                }
            }
            recombineDatas.add(recombineData1);
        }
        InfluxDbUtils influxDB = InfluxDbUtils.setUp();
        List<String> records = new ArrayList<>();
        for (RecombineData recombineData : recombineDatas) {
            if (!StringUtils.isEmpty(recombineData.getValue1().trim())) {
                try {
                    Map<String, String> tags = new HashMap<>();
                    tags.put("point", recombineData.getPointId());
                    Map<String, Object> values = new HashMap<>();
                    values.put("getway", recombineData.getGatewayNo().trim());
                    values.put("value1", Double.valueOf(Double.parseDouble(recombineData.getValue1().toString().trim())));
                    values.put("value2", Double.valueOf(Double.parseDouble(recombineData.getValue2().toString().trim())));
                    Point point = influxDB.pointBuilder(measurement, tags, values, recombineData.getTime().toString());
                    BatchPoints batchPoint = BatchPoints.database(measurement).consistency(InfluxDB.ConsistencyLevel.ALL).build();
                    batchPoint.point(point);
                    records.add(batchPoint.lineProtocol());
                } catch (Exception e) {
                    log.error(e.getMessage());
                }
            }
        }
        influxDB.batchInsert("data", null, InfluxDB.ConsistencyLevel.ALL, records);
    }

}